// Copyright (c) 2020-present,  INSPUR Co, Ltd.  All rights reserved.
// This source code is licensed under Apache 2.0 License.

#pragma once
#include <condition_variable>
#include <iostream>
#include <memory>
#include <mutex>
#include <queue>
namespace rocksdb {
template <typename T>
class ThreadSafeQueue {
 public:
  ThreadSafeQueue() = default;

  ~ThreadSafeQueue() {
  };

  void Push(T new_data) {
    std::lock_guard<std::mutex> lk(mut_);   // 1.Global lock
    data_queue_.push(std::move(new_data));  // 2.Exclusive lock when push
    cond_.notify_one();
  }

  void Clear() {
    std::lock_guard<std::mutex> lk(mut_);
    std::queue<T> empty;
    std::swap(empty, data_queue_);
  }

  void WaitAndPop(T &val) {
    std::unique_lock<std::mutex> ulk(mut_);  // 3.Global lock
    cond_.wait(ulk, [this]() {
      return !data_queue_.empty();
    });  // 4.Exclusive lock for front and pop_front
    val = std::move(data_queue_.front());
    data_queue_.pop();
  }

  std::shared_ptr<T> wait_and_pop() {
    std::unique_lock<std::mutex> ulk(mut_);
    cond_.wait(ulk, [this]() { return !data_queue_.empty(); });
    std::shared_ptr<T> val(std::make_shared<T>(std::move(data_queue_.front())));
    data_queue_.pop();
    return val;
  }

  bool TryPop(T &val) {
    std::lock_guard<std::mutex> lk(mut_);
    if (data_queue_.empty()) return false;
    val = std::move(data_queue_.front());
    data_queue_.pop();
    return true;
  }

  std::shared_ptr<T> try_pop() {
    std::shared_ptr<T> val;
    std::lock_guard<std::mutex> lk(mut_);
    if (data_queue_.empty()) return val;
    val = std::make_shared<T>(std::move(data_queue_.front()));
    data_queue_.pop();
    return val;
  }

  bool Empty() {
    std::lock_guard<std::mutex> lk(mut_);
    return data_queue_.empty();
  }

  size_t Size() const { return data_queue_.size(); }

 private:
  std::queue<T> data_queue_;
  std::mutex mut_;
  std::condition_variable cond_;
};
}  // namespace rocksdb